package jtcl.tech.kafka.factory;/**
 * @author sjt
 * @since 2021-08-13 13:57
 */

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.Deserializer;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.lang.Nullable;
import org.springframework.util.StringUtils;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

/**
 * <p></p>
 *
 * @author shijinting
 * @since 2021-08-13 13:57
 */
public class KafkaConsumerFactory<K, V> extends DefaultKafkaConsumerFactory {

    private Map<String, Object> configs;

    public Map<String, Object> getConfigs() {
        return configs;
    }

    public void setConfigs(Map<String, Object> configs) {
        this.configs = configs;
    }

    /**
     * Construct a factory with the provided configuration.
     *
     * @param configs the configuration.
     */
    public KafkaConsumerFactory(Map configs) {
        super(configs);
        this.configs=configs;
    }

    /**
     * Construct a factory with the provided configuration and deserializers.
     *
     * @param configs           the configuration.
     * @param keyDeserializer   the key {@link Deserializer}.
     * @param valueDeserializer the value {@link Deserializer}.
     */
    public KafkaConsumerFactory(Map configs, Deserializer keyDeserializer, Deserializer valueDeserializer) {
        super(configs, keyDeserializer, valueDeserializer);
        this.configs=configs;
    }

    @Override
    public Consumer<K, V> createConsumer(@Nullable String groupId, @Nullable String clientIdPrefix,
                                         @Nullable final String clientIdSuffixArg, @Nullable Properties properties) {

        return createKafkaConsumer(groupId, clientIdPrefix, clientIdSuffixArg, properties);
    }

    @Override
    protected Consumer<K, V> createKafkaConsumer(@Nullable String groupId, @Nullable String clientIdPrefix,
                                                      @Nullable final String clientIdSuffixArg, @Nullable Properties properties) {

        boolean overrideClientIdPrefix = StringUtils.hasText(clientIdPrefix);
        String clientIdSuffix = clientIdSuffixArg;
        if (clientIdSuffix == null) {
            clientIdSuffix = "";
        }
        boolean shouldModifyClientId = (this.configs.containsKey(ConsumerConfig.CLIENT_ID_CONFIG)
                && StringUtils.hasText(clientIdSuffix)) || overrideClientIdPrefix;
        if (groupId == null
                && (properties == null || properties.stringPropertyNames().size() == 0)
                && !shouldModifyClientId) {
            return createKafkaConsumer(this.configs);
        }
        else {
            return createConsumerWithAdjustedProperties(groupId, clientIdPrefix, properties, overrideClientIdPrefix,
                    clientIdSuffix, shouldModifyClientId);
        }
    }
    private Consumer<K, V> createConsumerWithAdjustedProperties(String groupId, String clientIdPrefix,
                                                                     Properties properties, boolean overrideClientIdPrefix, String clientIdSuffix,
                                                                     boolean shouldModifyClientId) {

        Map<String, Object> modifiedConfigs = new HashMap<>(this.configs);
        if (groupId != null) {
            modifiedConfigs.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        }
        if (shouldModifyClientId) {
            modifiedConfigs.put(ConsumerConfig.CLIENT_ID_CONFIG,
                    (overrideClientIdPrefix ? clientIdPrefix
                            : modifiedConfigs.get(ConsumerConfig.CLIENT_ID_CONFIG)) + clientIdSuffix);
        }
        if (properties != null) {
            properties.stringPropertyNames()
                    .stream()
                    .filter(name -> !name.equals(ConsumerConfig.CLIENT_ID_CONFIG)
                            && !name.equals(ConsumerConfig.GROUP_ID_CONFIG))
                    .forEach(name -> {
                        System.out.println(name);
                        System.out.println(properties.getProperty(name));
                        modifiedConfigs.put(name, properties.getProperty(name));
                    });
        }
        return super.createKafkaConsumer(modifiedConfigs);
    }
}
